package com.github.kezhenxu94.playground.java.kafka;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

public class DemoProducer {

	KafkaProducer<Integer, String> producer;
	Properties properties;

	public DemoProducer() throws Exception {
		properties = new Properties();
		properties.setProperty("bootstrap.servers", "localhost:9092");
		properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.IntegerSerializer");
		properties.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
		producer = new KafkaProducer<>(properties);
		BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
		for (String line = reader.readLine(); !"bye".equals(line); line = reader.readLine()) {
			producer.send(new ProducerRecord<Integer, String>("test", line));
		}
		producer.close();
	}

	public static void main(String[] args) throws Exception {
		new DemoProducer();
	}
}
